package com.qch.kafka.demo.producer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.function.Consumer;

/**
 * @Author qch
 * @Date 2021/5/26
 */
public class MyConsumer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,IntegerDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG,"consumer_demo");
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");


        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(configs);
        //先订阅再消费
        consumer.subscribe(Arrays.asList("topic_3"));
        //如果主题中没有可以消费的消息，则该方法可以放到while循环中，每过3秒重新拉去一次，
        // 如果还没有拉取到，过3秒再次拉去，防止while循环太密集的poll调用

        //批量从主题的分区中拉去消息
        ConsumerRecords<Integer, String> consumerRecords = consumer.poll(3_000);
        consumerRecords.forEach(new Consumer<ConsumerRecord<Integer, String>>() {
            @Override
            public void accept(ConsumerRecord<Integer, String> record) {
                System.out.println(record.topic()+"\t"+record.offset()+"\t"+record.key()+"\t"+record.value());
            }
        });
        consumer.close();
    }
}
